package com.awesome.spring;

import com.awesome.spring.function.process.BehaviorLogProcessFunc;
import com.awesome.spring.function.sink.BehaviorLogSinkFunc;
import com.awesome.spring.function.source.BehaviorLogSourceFunc;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Objects;
import java.util.Properties;

/**
 * @program: awesome-spring
 * @description: flinkjob
 * @author: Alex Wu
 * @createDate: 2025-03-20 15:05
 **/
public class RichFlinkKafkaJob {

    private static final String ENV_KAFKA_SERVER = "KAFKA_SERVER";
    private static final String KAFKA_BEHAVIOR_LOG_TOPIC = "test";
    private static final String KAFKA_BEHAVIOR_LOG_GROUP_ID = "test_group";
    private static final String JOB_NAME_BEHAVIOR_LOG = "Flink_Behavior_Log";
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            // 构建 KafkaSource
            KafkaSource<String> source = KafkaSource.<String>builder()
                    .setBootstrapServers("192.168.102.8:9092")
                    .setTopics(KAFKA_BEHAVIOR_LOG_TOPIC)
                    .setGroupId(KAFKA_BEHAVIOR_LOG_GROUP_ID)
                    .setStartingOffsets(OffsetsInitializer.earliest())
                    .setValueOnlyDeserializer(new SimpleStringSchema())
                    .setProperty("enable.auto.commit", "true")
                    .build();

            // 读取 Kafka 数据流
            DataStream<String> kafkaStream = env.fromSource(source,
                    WatermarkStrategy.noWatermarks(), "SOURCE_KAFKA");

            // 解析 Kafka 数据为 BehaviorLog 对象
            DataStream<String> mapStream = kafkaStream.map(new BehaviorLogSourceFunc()).name("KAFKA_MAP");

            // 中间处理结果，可选
            SingleOutputStreamOperator<String> processHttp = mapStream.process(new BehaviorLogProcessFunc()).name("PROCESS_HTTP");

            // 将数据写入 Doris
            processHttp.addSink(new BehaviorLogSinkFunc()).name("SINK_DORIS");

            // 启动 Flink 任务
            env.execute(JOB_NAME_BEHAVIOR_LOG);
        }
}